package com.rabbit.mq.service;

import com.rabbit.mq.enums.BindingEnum;
import com.rabbit.mq.enums.ExchangeEnum;
import com.rabbit.mq.enums.ExchangeTypeEnum;
import com.rabbit.mq.enums.QueueEnum;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.AsyncRabbitTemplate;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.BatchingRabbitTemplate;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.util.UUID;

@Service
@Slf4j
public class RabbitmqService {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Autowired
    private AsyncRabbitTemplate asyncRabbitTemplate;
    @Autowired
    private RabbitPublisher rabbitPublisher;

    //发送消息方法调用: 构建Message消息
    public void send() throws Exception {
        // 消息属性
        MessageProperties properties = getMessageProperties();
        // 遍历交换机枚举
        QueueEnum.toList().forEach(queueEnum -> {
            // 执行发布信息
            String message = queueEnum.getName().concat("_message");
            ExchangeEnum exchangeEnum = BindingEnum.getExchangeEnum(queueEnum);
            // 未绑定交换机
            if (null == exchangeEnum){
                // 简单模式 -- 队列是direct模式，且没有跟交换机绑定，直接使用RoutingKey访问
                // 这种情况下，RoutingKey与队列名称相同
                log.info("---------------" + queueEnum.getName() + "--------------start---------");
                rabbitTemplate.convertSendAndReceive(
                        // RoutingKey与队列名称相同
                        queueEnum.getName().concat("_queue"),
                        new Message(message.getBytes(), properties));
                log.info("---------------" + queueEnum.getName() + "--------------end---------");
                return;
            }
            // 广播模式不需要指定routingKey,全部发送
            if (exchangeEnum.getType().equals(ExchangeTypeEnum.fanout)){
                log.info("---------------" + queueEnum.getName() + "--------------start---------");
                rabbitTemplate.convertSendAndReceive(
                        exchangeEnum.getExchangeName() + "_exchange",
                        "",
                        new Message(message.getBytes(), properties));
                log.info("---------------" + queueEnum.getName() + "--------------end---------");
                return;
            }
            // 其他 -- 根据交换机、routing直接匹配
            log.info("---------------" + queueEnum.getName() + "--------------start---------");
            rabbitTemplate.convertSendAndReceive(
                    exchangeEnum.getExchangeName() + "_exchange",
                    queueEnum.getRoutingKey(),
                    new Message(message.getBytes(), properties));
            log.info("---------------" + queueEnum.getName() + "---------------end--------");
        });
    }

    /**
     * 延迟队列测试
     */
    public void deal_queue_test() {
        ExchangeEnum exchangeEnum = BindingEnum.deal_binding.getExchangeEnum();
        QueueEnum queueEnum = BindingEnum.deal_binding.getQueueEnum();
        // 消息
        String message = "11111111111111111111111111111111111111";
        MessageProperties messageProperties = getMessageProperties();
        // 发送
        rabbitTemplate.convertSendAndReceive(
                exchangeEnum.getExchangeName(),
                queueEnum.getRoutingKey(),
                new Message(message.getBytes(), messageProperties));
    }

    /**
     * 消息属性
     * @return
     */
    public MessageProperties getMessageProperties(){
        MessageProperties properties = new MessageProperties();
        //消息内容的编码格式
        properties.setContentEncoding("UTF-8");
        //Delivery mode: 是否持久化，1 - Non-persistent，2 - Persistent
        properties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
        // 设置超时时间--有效
        properties.setExpiration("5000");
        return properties;
    }

    public void async() {
        log.info("---------------async--------------start---------");
        ExchangeEnum exchangeEnum = BindingEnum.async_binding.getExchangeEnum();
        QueueEnum queueEnum = BindingEnum.async_binding.getQueueEnum();
        AsyncRabbitTemplate.RabbitConverterFuture<Object> future =
                asyncRabbitTemplate.convertSendAndReceive(exchangeEnum.getExchangeName(), queueEnum.getRoutingKey(), "123123123");
        // 回调监听 配置下面代码时 如果 队列监听中没有返回值时会报错
        future.addCallback(new ListenableFutureCallback<Object>() {
            @Override
            public void onFailure(Throwable ex) {
                ex.printStackTrace();
            }
            @Override
            public void onSuccess(Object result) {
                System.out.println("回调收到结果=> " + result);
            }
        });
        log.info("---------------async--------------end---------");
    }

    public void sync() throws Exception {
        log.info("---------------sync--------------start---------");
        rabbitTemplate.convertSendAndReceive("reply_exchange", "reply.queue", "2222");
        log.info("---------------sync--------------end---------");
    }

    public void publisher() throws Exception {
        log.info("---------------sync--------------start---------");
        rabbitPublisher.publish("reply_exchange", "reply.queue", "2222");
        log.info("---------------sync--------------end---------");
    }

    /**
     * 唯一标识
     * @return
     */
    public CorrelationData getCorrelationData(){
        CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
        return correlationData;
    }


    @Autowired
    private BatchingRabbitTemplate batchingRabbitTemplate;

    public void batch() {
        String msg;
        Message message;
        MessageProperties messageProperties=new MessageProperties();
        long timestamp=System.currentTimeMillis();
        ExchangeEnum exchangeEnum = BindingEnum.batch_binding.getExchangeEnum();
        QueueEnum queueEnum = BindingEnum.batch_binding.getQueueEnum();
        log.info("---------------batch--------------start---------");
        for(int i=0;i<9;i++){
            msg="batch."+timestamp+"-"+i;

            message=new Message(msg.getBytes(), messageProperties);
            batchingRabbitTemplate.convertAndSend(exchangeEnum.getExchangeName(), queueEnum.getRoutingKey(), message);
        }
        log.info("---------------batch--------------end---------");
    }
}
